package com.yunhe.history.websocket;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.yunhe.common.constant.KafkaConsts;
import com.yunhe.common.constant.MeasurementConsts;
import com.yunhe.common.constant.SystemConsts;
import com.yunhe.common.model.data.Measurement;
import com.yunhe.common.model.msg.MeasurementMsgDTO;
import com.yunhe.common.util.CollectionUtil;
import com.yunhe.common.websocket.WebsocketIncrementHandler;
import com.yunhe.history.service.data.MeasurementService;
import com.yunhe.common.util.StringUtil;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 历史量测曲线WebSocket处理类
 * @author liuronglei
 */
@EnableAsync
public class MeasurementIncrementHandler extends WebsocketIncrementHandler {
    private static final ObjectMapper mapper = new ObjectMapper();
    @Autowired
    private MeasurementService measurementService;

    @Override
    public void beforeSend(WebSocketSession session, Object content) {

    }

    @Override
    public void afterEstablished(WebSocketSession session) {

    }

    @Override
    public void afterClosed(WebSocketSession session, CloseStatus status) {

    }

    @Override
    public void handleMessage(WebSocketSession session, TextMessage message, List<String> points, String startDate, String frequency) {
        if (CollectionUtil.isNotEmpty(points)) {
            Map<Object, List<Measurement>> result = new HashMap<>();
            for (String point : points) {
                if (!StringUtil.isEmpty(startDate)) {
                    List<Measurement> measurements = (List<Measurement>)measurementService.findMeasurementsByStartDate(Long.parseLong(point),
                            startDate, frequency, null);
                    result.put(point, measurements);
                } else {
                    List<Measurement> measurements = new ArrayList<>();
                    Measurement measurement = measurementService.findMeasurementsLast(Long.parseLong(point));
                    measurements.add(measurement);
                    result.put(point, measurements);
                }
            }
            sendMessage(result, frequency, session, points);
        }
    }

    @KafkaListener(topics = KafkaConsts.Topic.MeasurementRefresh, groupId = KafkaConsts.Group.RandomGroupId)
    public void measurementListener(ConsumerRecord<String, String> record) throws Exception {
        String value = record.value();
        if (value != null) {
            MeasurementMsgDTO message = mapper.readValue(value, MeasurementMsgDTO.class);
            String action = message.getAction();
            if (SystemConsts.ACTION.ADD.equals(action)) {
                List<Measurement> list = message.getMeasurements();
                String frequency = message.getFrequency();
                frequency = StringUtil.isEmpty(frequency) ? MeasurementConsts.FREQUENCY.Minute15.value() : frequency;
                Map<String, List<Measurement>> result = new HashMap<>();
                if (list != null && list.size() > 0) {
                    for (Measurement measurement : list) {
                        Long pointNumber = measurement.getPointNumber();
                        String pid = String.valueOf(pointNumber);
                        List<Measurement> values = result.computeIfAbsent(pid, v -> new ArrayList<>());
                        values.add(measurement);
                    }
                    sendMessage(result, frequency, null, new ArrayList<>(result.keySet()));
                }
            }
        }
    }
}